package cn.tedu;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

/**
 * @author Amos
 * @date 2022/5/13
 */

public class MyProducer {
    public static void main(String[] args) throws InterruptedException {
        // 创建kafka属性配置对象
        Properties props = new Properties();

        // 连接服务器地址
        props.put("bootstrap.servers", "hadoop01:9092,hadoop02:9092,hadoop03:9092");
        /*
        ack 确认：
        1) 0: 表示不管分区是否存储数据，直接发送下一条生产数据
        2) 1: 表示leader保存上数据后，再发送下一条生产数据
        3) -1(all): 表示leader和follower都要保存好数据后，再发送下一条生产数据
         */
        props.put("acks", "all");
        // 发送数据的重试次数
        props.put("retries", 0);
        // 批处理发送数据大小16k
        props.put("batch.size", 16384);
        // 缓存大小32m
        props.put("buffer.memory", 33554432);
        // 延时发送数据1ms
        props.put("linger.ms", 1);
        // key和value的序列化
        props.put("key.serializer",
                "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer",
                "org.apache.kafka.common.serialization.StringSerializer");

        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props);

        for (int i = 0; i < 1000; i++) {
            producer.send(new ProducerRecord<String, String>("test", "生产数据：" + i));
            Thread.sleep(500);
        }
        // 格式gitee
    }
}
